"""Flink实时流处理模块，用于实时处理金融数据流"""

import json
import os
import sys
from datetime import datetime
from typing import Optional, Dict, Any, Tuple

from fin_senti_entity_platform.utils.config_loader import config_loader
from fin_senti_entity_platform.utils.logger import Logger

# 直接导入必要的Flink组件
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors import FlinkKafkaConsumer, FlinkKafkaProducer
from pyflink.common.serialization import SimpleStringSchema
from pyflink.common.typeinfo import Types

# 初始化日志记录器
logger = Logger.get_logger('flink_stream_processor', 'flink_processing.log')

class FlinkStreamProcessor:
    """
    Flink实时流处理器，用于实时处理金融数据流
    """
    def __init__(self):
        """
        初始化Flink流处理器
        """
        # 加载Flink配置
        self.flink_config = config_loader.config.get('data_collection', {}).get('flink', {})

        # 加载Kafka配置
        self.kafka_config = config_loader.config.get('data_collection', {}).get('kafka', {})
        
        # 初始化执行环境
        self.env: Optional[StreamExecutionEnvironment] = None
        self.stream = None
        
        # 输入输出主题
        self.input_topic = self.flink_config.get('input_topic', 'financial_news_raw')
        self.output_topic = self.flink_config.get('output_topic', 'financial_news_processed')
        self.error_topic = self.flink_config.get('error_topic', 'financial_news_error')
        
        # 检查平台兼容性
        self._check_platform_compatibility()
        
        logger.info("Flink流处理器初始化完成")
        
    def _check_platform_compatibility(self):
        """
        检查平台兼容性并应用特殊配置
        """
        if sys.platform == 'win32':
            logger.info("检测到Windows平台，应用Windows特定配置")
            # Windows平台可能需要特殊配置
            # 设置环境变量以优化Windows上的性能
            os.environ.setdefault('HADOOP_HOME', 'C:\\winutils')
            os.environ.setdefault('JAVA_HOME', os.environ.get('JAVA_HOME', ''))
        
    def init_environment(self) -> bool:
        """
        初始化Flink执行环境
        :return: 是否初始化成功
        """
        try:
            # 创建执行环境
            self.env = StreamExecutionEnvironment.get_execution_environment()
            
            # 配置并行度
            parallelism = self.flink_config.get('parallelism', 4)
            self.env.set_parallelism(parallelism)
            
            # 配置检查点
            checkpoint_interval = self.flink_config.get('checkpoint_interval', 60000)  # 默认60秒
            if checkpoint_interval > 0:
                self.env.enable_checkpointing(checkpoint_interval)
                # 配置检查点模式和超时时间
                self.env.get_checkpoint_config().set_checkpointing_mode('EXACTLY_ONCE')
                self.env.get_checkpoint_config().set_checkpoint_timeout(120000)
                
            # Windows平台特殊配置
            if sys.platform == 'win32':
                # 禁用JVM堆外内存分配，提高Windows兼容性
                self.env.get_config().set_use_deprecated_java_gateway(True)
                
            logger.info(f"Flink执行环境初始化完成，并行度: {parallelism}")
            return True
            
        except Exception as e:
            logger.error(f"初始化Flink执行环境失败: {str(e)}")
            # 提供更详细的错误信息和解决方案
            if 'java.lang' in str(e) or 'JVM' in str(e).upper():
                logger.error("这可能是Java环境配置问题。请确保正确安装了Java并设置了JAVA_HOME环境变量。")
            return False
            
    def create_kafka_consumer(self) -> Optional[FlinkKafkaConsumer]:
        """
        创建Kafka消费者
        :return: Kafka消费者实例
        """
        try:
            # Kafka消费者配置
            kafka_props = {
                'bootstrap.servers': self.kafka_config.get('bootstrap_servers', 'localhost:9092'),
                'group.id': self.kafka_config.get('consumer_group', 'flink_consumer_group'),
                'auto.offset.reset': self.kafka_config.get('auto_offset_reset', 'latest')
            }
            
            # 添加额外的Kafka配置参数
            if self.kafka_config.get('security_protocol'):
                kafka_props['security.protocol'] = self.kafka_config.get('security_protocol')
            if self.kafka_config.get('sasl_mechanism'):
                kafka_props['sasl.mechanism'] = self.kafka_config.get('sasl_mechanism')
            if self.kafka_config.get('sasl_jaas_config'):
                kafka_props['sasl.jaas.config'] = self.kafka_config.get('sasl_jaas_config')
                
            # 创建Kafka消费者
            consumer = FlinkKafkaConsumer(
                topics=self.input_topic,
                deserialization_schema=SimpleStringSchema(),
                properties=kafka_props
            )
            
            logger.info(f"Kafka消费者创建完成，主题: {self.input_topic}")
            return consumer
            
        except Exception as e:
            logger.error(f"创建Kafka消费者失败: {str(e)}")
            return None
            
    def create_kafka_producer(self, topic: str) -> Optional[FlinkKafkaProducer]:
        """
        创建Kafka生产者
        :param topic: 输出主题
        :return: Kafka生产者实例
        """
        try:
            # Kafka生产者配置
            kafka_props = {
                'bootstrap.servers': self.kafka_config.get('bootstrap_servers', 'localhost:9092')
            }
            
            # 添加额外的Kafka配置参数
            if self.kafka_config.get('acks'):
                kafka_props['acks'] = self.kafka_config.get('acks')
            if self.kafka_config.get('retries'):
                kafka_props['retries'] = self.kafka_config.get('retries')
            
            # 创建Kafka生产者
            producer = FlinkKafkaProducer(
                topic=topic,
                serialization_schema=SimpleStringSchema(),
                properties=kafka_props
            )
            
            logger.info(f"Kafka生产者创建完成，主题: {topic}")
            return producer
            
        except Exception as e:
            logger.error(f"创建Kafka生产者失败: {str(e)}")
            return None
            
    def build_stream(self) -> bool:
        """
        构建数据处理流
        :return: 是否构建成功
        """
        try:
            if self.env is None:
                if not self.init_environment():
                    return False
                    
            # 创建Kafka消费者
            consumer = self.create_kafka_consumer()
            if consumer is None:
                return False
                
            # 从Kafka读取数据流
            self.stream = self.env.add_source(consumer)
            
            # 解析JSON数据
            self.stream = self.stream.map(
                self._parse_json_data,
                output_type=Types.TUPLE([Types.STRING(), Types.MAP(Types.STRING(), Types.STRING())])
            )
            
            # 过滤有效的数据
            self.stream = self.stream.filter(self._filter_valid_data)
            
            # 实时处理数据
            self.stream = self.stream.map(self._process_data)
            
            # 创建输出生产者
            output_producer = self.create_kafka_producer(self.output_topic)
            error_producer = self.create_kafka_producer(self.error_topic)
            
            if output_producer is None or error_producer is None:
                return False
                
            # 分流处理结果
            self.stream.map(
                lambda x: json.dumps(x['data']) if x['status'] == 'success' else None
            ).filter(lambda x: x is not None).add_sink(output_producer)
            
            self.stream.map(
                lambda x: json.dumps(x['error_info']) if x['status'] == 'error' else None
            ).filter(lambda x: x is not None).add_sink(error_producer)
            
            logger.info("数据处理流构建完成")
            return True
            
        except Exception as e:
            logger.error(f"构建数据处理流失败: {str(e)}")
            return False
            
    def _parse_json_data(self, json_str: str) -> Tuple[str, Dict[str, Any]]:
        """
        解析JSON字符串
        :param json_str: JSON字符串
        :return: 解析后的数据元组 (原始字符串, 解析后的数据)
        """
        try:
            data = json.loads(json_str)
            return (json_str, data)
        except Exception as e:
            logger.warning(f"JSON解析失败: {str(e)}, 原始数据: {json_str[:100]}...")
            # 解析失败，返回原始字符串和空数据
            return (json_str, {})
            
    def _filter_valid_data(self, data_tuple: Tuple[str, Dict[str, Any]]) -> bool:
        """
        过滤有效的数据
        :param data_tuple: 数据元组 (原始字符串, 解析后的数据)
        :return: 是否有效
        """
        _, data = data_tuple
        
        # 检查必要字段
        required_fields = ['content', 'title', 'url']
        for field in required_fields:
            if field not in data or not data[field]:
                return False
                
        # 检查内容长度
        content_length = len(data.get('content', ''))
        if content_length < 50:
            return False
            
        return True
            
    def _process_data(self, data_tuple: Tuple[str, Dict[str, Any]]) -> Dict[str, Any]:
        """
        处理单条数据
        :param data_tuple: 数据元组 (原始字符串, 解析后的数据)
        :return: 处理结果
        """
        raw_str, data = data_tuple
        
        try:
            # 这里实现数据处理逻辑
            # 实际项目中，这里会调用数据清洗器等组件
            
            # 添加处理时间
            data['processed_time'] = datetime.now().isoformat()
            
            # 添加处理状态
            data['process_status'] = 'success'
            
            # 返回处理成功的结果
            return {
                'status': 'success',
                'data': data,
                'error_info': None
            }
            
        except Exception as e:
            # 处理失败，返回错误信息
            error_info = {
                'raw_data': raw_str,
                'error_message': str(e),
                'error_time': datetime.now().isoformat()
            }
            
            logger.error(f"数据处理失败: {str(e)}")
            return {
                'status': 'error',
                'data': None,
                'error_info': error_info
            }
            
    def add_window_operation(self, window_size: int = 60, slide_size: int = 30) -> bool:
        """
        添加窗口操作
        :param window_size: 窗口大小（秒）
        :param slide_size: 滑动大小（秒）
        :return: 是否添加成功
        """
        try:
            if self.stream is None:
                logger.error("数据流未初始化，请先调用build_stream方法")
                return False
                
            # 实现窗口操作逻辑
            # 这里可以根据实际需求添加各种窗口操作
            
            logger.info(f"窗口操作添加完成，窗口大小: {window_size}s, 滑动大小: {slide_size}s")
            return True
            
        except Exception as e:
            logger.error(f"添加窗口操作失败: {str(e)}")
            return False
            
    def start(self) -> bool:
        """
        启动Flink流处理
        :return: 是否启动成功
        """
        try:
            if self.stream is None:
                if not self.build_stream():
                    return False
                    
            # 启动执行
            logger.info("Flink流处理开始执行")
            self.env.execute("Financial News Stream Processing")
            
            return True
            
        except Exception as e:
            logger.error(f"启动Flink流处理失败: {str(e)}")
            # 提供更详细的错误信息和解决方案
            if 'JVM' in str(e).upper():
                logger.error("这可能是Java环境配置问题。请确保正确安装了Java并设置了JAVA_HOME环境变量。")
            elif 'Connection refused' in str(e):
                logger.error("无法连接到Kafka服务。请检查Kafka服务是否正常运行以及配置是否正确。")
            return False
            
    def stop(self) -> bool:
        """
        停止Flink流处理
        :return: 是否停止成功
        """
        # Flink作业通常是长期运行的，这里可以添加停止逻辑
        # 实际项目中，可能需要通过Flink的REST API或其他方式停止作业
        logger.info("Flink流处理已停止")
        return True